package avicit.bdp.dds.api.service;

import avicit.bdp.common.utils.BdpLogUtil;
import avicit.bdp.common.utils.DateUtils;
import avicit.bdp.common.utils.redis.RedisCacheHelper;
import avicit.bdp.core.util.json.JSONUtils;
import avicit.bdp.dds.api.enums.ExecuteType;
import avicit.bdp.dds.api.enums.ProcessType;
import avicit.bdp.dds.api.enums.Status;
import avicit.bdp.dds.common.Constants;
import avicit.bdp.dds.dao.entity.Command;
import avicit.bdp.dds.dao.entity.ProcessDefinition;
import avicit.bdp.dds.dao.entity.ProcessInstance;
import avicit.bdp.dds.dao.entity.Schedule;
import avicit.bdp.dds.dao.mapper.ProcessDefinitionMapper;
import avicit.bdp.dds.dispatch.enums.CommandType;
import avicit.bdp.dds.dispatch.enums.ExecutionStatus;
import avicit.bdp.dds.dispatch.enums.ReleaseState;
import avicit.bdp.dds.dispatch.enums.RunMode;
import avicit.bdp.dds.dispatch.enums.TaskType;
import avicit.bdp.dds.dispatch.model.Server;
import avicit.bdp.dds.service.process.ProcessService;
import avicit.bdp.dds.service.quartz.MonitorDirectoryScheduleJob;
import avicit.bdp.dds.service.quartz.QuartzExecutors;
import avicit.bdp.dds.service.quartz.cron.CronUtils;
import avicit.platform6.core.exception.BusinessException;
import avicit.platform6.core.properties.PlatformConstant;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.text.ParseException;
import java.util.Date;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;

import static avicit.bdp.dds.common.Constants.CMDPARAM_COMPLEMENT_DATA_END_DATE;
import static avicit.bdp.dds.common.Constants.CMDPARAM_COMPLEMENT_DATA_START_DATE;
import static avicit.bdp.dds.common.Constants.CMDPARAM_RECOVER_PROCESS_ID_STRING;
import static avicit.bdp.dds.common.Constants.CMDPARAM_START_NODE_NAMES;
import static avicit.bdp.dds.common.Constants.COMMA;
import static avicit.bdp.dds.common.Constants.MAX_TASK_TIMEOUT;

/**
 * @author mayj
 * 流程实例执行
 */
@Service
public class ExecutorService {
    private static final Logger logger = LoggerFactory.getLogger(ExecutorService.class);

    @Autowired
    private MonitorService monitorService;
    @Autowired
    private ProcessDefinitionMapper processDefinitionMapper;
    @Autowired
    private ProcessService processService;
    @Autowired
    private CommandService commandService;
    @Autowired
    private ProcessInstanceService processInstanceService;
    @Autowired
    private SchedulerService schedulerService;


    /**
     * 检查master服务是否存在
     *
     * @return master服务是否存在
     */
    private boolean checkMasterExists() {
        List<Server> masterServers = monitorService.getServerListFromZK(true);
        if (masterServers.size() == 0) {
            throw new BusinessException(Status.MASTER_NOT_EXISTS.getMsg());
        }

        return true;
    }

    /**
     * 检查工作流合法性；（1）工作流是否存在（2）工作流是否处于ONLINE状态
     *
     * @param processDefinition   processDefinition
     * @param processDefinitionId processDefinitionId
     * @return 工作流合法性
     */
    private boolean checkProcessDefinitionValid(ProcessDefinition processDefinition, String processDefinitionId) {
        if (processDefinition == null) {
            logger.error("流程定义{}不存在", processDefinitionId);
            return false;
        } else if (processDefinition.getReleaseState() != ReleaseState.ONLINE.getCode()) {
            logger.error("流程定义{}处于OFFLINE状态", processDefinition.getId());
            return false;
        }

        return true;
    }

    /**
     * 检查执行类型合法性
     *
     * @param processInstance processInstance
     * @param executeType     executeType
     * @return 执行类型合法性
     */
    private boolean checkExecuteType(ProcessInstance processInstance, ExecuteType executeType) {
        ExecutionStatus executionStatus = ExecutionStatus.of(processInstance.getState());

        boolean checkResult = false;
        switch (executeType) {
            case PAUSE:
            case STOP:
                if (executionStatus.typeIsRunning()) {
                    checkResult = true;
                }
                break;
            case REPEAT_RUNNING:
                if (executionStatus.typeIsFinished()) {
                    checkResult = true;
                }
                break;
            case START_FAILURE_TASK_PROCESS:
                if (executionStatus.typeIsFailure()) {
                    checkResult = true;
                }
                break;
            case RECOVER_SUSPENDED_PROCESS:
                if (executionStatus.typeIsPause() || executionStatus.typeIsCancel()) {
                    checkResult = true;
                }
                break;
            default:
                break;
        }
        if (!checkResult) {
            logger.error("流程实例{}状态{}不允许执行{}操作", processInstance.getName(), executionStatus.toString(), executeType.toString());
        }

        return checkResult;
    }

    /**
     * 更新流程实例
     *
     * @param processInstance processInstance
     * @param commandType     commandType
     * @param executionStatus executionStatus
     * @return 更新流程实例结果
     */
    private boolean updateProcessInstancePrepare(ProcessInstance processInstance,
                                                 Integer commandType,
                                                 Integer executionStatus) {

        logger.info("update process instance [{}] prepare, commandType: {}, executionStatus: {}",
                processInstance.getId(), commandType, executionStatus);

        processInstance.setCommandType(commandType);
        processInstance.addHistoryCmd(CommandType.of(commandType));
        processInstance.setState(executionStatus);

        int update = processInstanceService.updateByPrimaryKey(processInstance);
        //记录日志
        String logTitle = CommandType.of(commandType).getDescp() + processInstance.getName() + "任务";
        BdpLogUtil.log("数据计算模块", logTitle, PlatformConstant.OpType.update);

        if (update <= 0) {
            logger.error("更新流程实例{}失败", processInstance.getId());
            return false;
        }

        return true;
    }

    /**
     * 检查是否需要生成CMD
     *
     * @param command command
     * @return 是否需要生成CMD
     */
    public Boolean verifyIsNeedCreateCommand(Command command) {
        boolean isNeedCreate = true;
        Map<Integer, Integer> cmdTypeMap = new HashMap<>(16);
        cmdTypeMap.put(CommandType.REPEAT_RUNNING.getCode(), 1);
        cmdTypeMap.put(CommandType.RECOVER_SUSPENDED_PROCESS.getCode(), 1);
        cmdTypeMap.put(CommandType.START_FAILURE_TASK_PROCESS.getCode(), 1);

        if (cmdTypeMap.containsKey(command.getCommandType())) {
            JSONObject cmdParamObj = (JSONObject) JSON.parse(command.getCommandParam());
            JSONObject tempObj;
            String processInstanceId = cmdParamObj.getString(CMDPARAM_RECOVER_PROCESS_ID_STRING);

            List<Command> commands = commandService.findCommandList();
            for (Command tmpCommand : commands) {
                if (cmdTypeMap.containsKey(tmpCommand.getCommandType())) {
                    tempObj = (JSONObject) JSON.parse(tmpCommand.getCommandParam());
                    if (tempObj != null && processInstanceId.equals(tempObj.getString(CMDPARAM_RECOVER_PROCESS_ID_STRING))) {
                        isNeedCreate = false;
                        break;
                    }
                }
            }
        }
        return isNeedCreate;
    }

    /**
     * 新增一条CMD
     *
     * @param loginUserId         loginUserId
     * @param instanceId          instanceId
     * @param processDefinitionId processDefinitionId
     * @param commandType         commandType
     * @return 新增结果
     */
    private boolean insertCommand(String loginUserId, String instanceId, String processDefinitionId, Integer commandType) {
        Command command = new Command();
        command.setCommandType(commandType);
        command.setProcessDefinitionId(processDefinitionId);
        command.setCommandParam(String.format("{\"%s\":\"%s\"}", CMDPARAM_RECOVER_PROCESS_ID_STRING, instanceId));
        command.setExecutorId(loginUserId);

        if (!verifyIsNeedCreateCommand(command)) {
            logger.error("流程实例{}正在创建CMD,请等待...", processDefinitionId);
            return false;
        }

        int create = processService.saveCommand(command);
        if (create <= 0) {
            logger.error("创建流程实例{}失败", instanceId);
            return false;
        }

        return true;
    }

    /**
     * 创建CMD
     *
     * @param commandType             commandType
     * @param processDefineId         processDefineId
     * @param nodeDep                 nodeDep
     * @param failureStrategy         failureStrategy
     * @param startNodeList           startNodeList
     * @param schedule                schedule
     * @param warningType             warningType
     * @param executorId              executorId
     * @param warningUserIds          warningUserIds
     * @param runMode                 runMode
     * @param processInstancePriority processInstancePriority
     * @param workerGroup             workerGroup
     * @return 插入结果
     * @throws ParseException parse exception
     */
    private Command createCommand(Integer commandType, String processDefineId, Integer nodeDep,
                                  Integer failureStrategy, String startNodeList, String schedule,
                                  Integer warningType, String executorId, String warningUserIds,
                                  Integer runMode, Integer processInstancePriority, String workerGroup) throws ParseException {
        Command command = new Command();

        Map<String, String> cmdParam = new HashMap<>(16);
        if (commandType == null) {
            command.setCommandType(CommandType.START_PROCESS.getCode());
        } else {
            command.setCommandType(commandType);
        }

        command.setProcessDefinitionId(processDefineId);
        if (nodeDep != null) {
            command.setTaskDependType(nodeDep);
        }

        if (failureStrategy != null) {
            command.setFailureStrategy(failureStrategy);
        }

        if (StringUtils.isNotEmpty(startNodeList)) {
            cmdParam.put(CMDPARAM_START_NODE_NAMES, startNodeList);
        }

        if (warningType != null) {
            command.setWarningType(warningType);
        }

        command.setCommandParam(JSONUtils.toJsonString(cmdParam));
        command.setExecutorId(executorId);
        command.setWarningUserIds(warningUserIds);
        command.setProcessInstancePriority(processInstancePriority);
        command.setStartTime(DateUtils.getCurrentDate());

        Date start = null;
        Date end = null;
        if (StringUtils.isNotEmpty(schedule)) {
            String[] interval = schedule.split(COMMA);
            if (interval.length == 2) {
                start = DateUtils.getScheduleDate(interval[0]);
                end = DateUtils.getScheduleDate(interval[1]);
            }
        }

        // determine whether to complement
        if (commandType == CommandType.COMPLEMENT_DATA.getCode()) {
            runMode = (runMode == null) ? RunMode.RUN_MODE_SERIAL.getCode() : runMode;
            if (null != start && null != end && !start.after(end)) {
                if (runMode == RunMode.RUN_MODE_SERIAL.getCode()) {
                    cmdParam.put(CMDPARAM_COMPLEMENT_DATA_START_DATE, DateUtils.dateToString(start));
                    cmdParam.put(CMDPARAM_COMPLEMENT_DATA_END_DATE, DateUtils.dateToString(end));
                    command.setCommandParam(JSONUtils.toJsonString(cmdParam));
                    processService.saveCommand(command);
                } else if (runMode == RunMode.RUN_MODE_PARALLEL.getCode()) {
                    List<Schedule> schedules = processService.queryReleaseSchedulerListByProcessDefinitionId(processDefineId);
                    List<Date> listDate = new LinkedList<>();
                    if (!CollectionUtils.isEmpty(schedules)) {
                        for (Schedule item : schedules) {
                            listDate.addAll(CronUtils.getSelfFireDateList(start, end, item.getCrontab()));
                        }
                    }
                    if (!CollectionUtils.isEmpty(listDate)) {
                        // loop by schedule date
                        for (Date date : listDate) {
                            cmdParam.put(CMDPARAM_COMPLEMENT_DATA_START_DATE, DateUtils.dateToString(date));
                            cmdParam.put(CMDPARAM_COMPLEMENT_DATA_END_DATE, DateUtils.dateToString(date));
                            command.setCommandParam(JSONUtils.toJsonString(cmdParam));
                            processService.saveCommand(command);
                        }
                        return command;
                    } else {
                        // loop by day
                        int runCunt = 0;
                        while (!start.after(end)) {
                            runCunt += 1;
                            cmdParam.put(CMDPARAM_COMPLEMENT_DATA_START_DATE, DateUtils.dateToString(start));
                            cmdParam.put(CMDPARAM_COMPLEMENT_DATA_END_DATE, DateUtils.dateToString(start));
                            command.setCommandParam(JSONUtils.toJsonString(cmdParam));
                            processService.saveCommand(command);
                            start = DateUtils.getSomeDay(start, 1);
                        }
                        return command;
                    }
                }
            } else {
                logger.error("there is not valid schedule date for the process definition: id:{},date:{}",
                        processDefineId, schedule);
            }
        } else {
            command.setCommandParam(JSONUtils.toJsonString(cmdParam));
            int count = processService.saveCommand(command);
            if (count > 0) {
                return command;
            }
        }

        return null;
    }

    /**
     * 启动工作流
     *
     * @param processDefinitionId     processDefinitionId
     * @param cronTime                cronTime
     * @param commandType             commandType
     * @param failureStrategy         failureStrategy
     * @param startNodeList           startNodeList
     * @param taskDependType          taskDependType
     * @param warningType             warningType
     * @param warningUserIds          warningUserIds
     * @param runMode                 runMode
     * @param processInstancePriority processInstancePriority
     * @param workerGroup             workerGroup
     * @param timeout                 timeout
     * @throws ParseException parse exception
     */
    public String execProcessInstance(String processDefinitionId, String cronTime,
                                      Integer commandType, Integer failureStrategy,
                                      String startNodeList, Integer taskDependType,
                                      Integer warningType, String warningUserIds,
                                      Integer runMode, Integer processInstancePriority,
                                      String workerGroup, Integer timeout) throws ParseException {
        // step1:检查超时时间合法性
        if (timeout <= 0 || timeout > MAX_TASK_TIMEOUT) {
            logger.error("任务超时时间不合法");
            throw new BusinessException("task timeout parameter is not valid");
        }

        // step2：检查流程定义合法性
        ProcessDefinition processDefinition = processDefinitionMapper.selectByPrimaryKey(processDefinitionId);
        if (processDefinition == null) {
            throw new BusinessException("流程定义不存在");
        }

        //检查资源配置是否配置
//        if (processDefinition.getType() == ProcessType.DATA_DEVELOP.getCode()) {
//            ProcessResourceSettings query = new ProcessResourceSettings();
//            query.setProcessDefinitionId(processDefinitionId);
//            //检查spark参数
//            if (processDefinition.getDatasourceId().startsWith(ProcessDefinition.DATASOURCE_TYPE_SPARK)) {
//                query.setType(ProcessResourceSettingsType.SPARK.getCode());
//                ProcessResourceSettings processResourceSettings = processResourceConfigService.selectOne(query);
//                if (processResourceSettings == null || StringUtils.isBlank(processResourceSettings.getEngineResourceId())) {
//                    throw new BusinessException("spark运行集群未设置，请进入【编辑-->资源配置】界面，设置spark当前集群！");
//                }
//            }
//        }

        //检查流程定义是否存在
        if (StringUtils.isBlank(processDefinition.getProcessDefinitionJson())) {
            throw new BusinessException("流程节点尚未定义，请先设计流程");
        }

        // step3:检查master服务是否存在
//        if (!checkMasterExists()) {
//            logger.error("master service doesn't exist");
//            throw new BusinessException("master服务不存在");
//        }
        //检查流程实例是否正在运行
        ProcessInstance processInstance = processInstanceService.queryLastOneProcess(processDefinitionId);
        if (processInstance != null && ExecutionStatus.of(processInstance.getState()).typeIsRunning()) {
            logger.error("流程实例正在运行中");
            throw new BusinessException("流程实例正在运行中，请运行结束后再执行！");
        }
        //处理任务通知
        if (processDefinition.getType() == ProcessType.DATA_DEVELOP.getCode()) {
            List<Schedule> schedules = schedulerService.queryByProcessDefinitionId(processDefinitionId);
            if (!schedules.isEmpty()) {
                Schedule schedule = schedules.get(0);
                failureStrategy = schedule.getFailureStrategy();
                warningType = schedule.getWarningType();
                warningUserIds = schedule.getWarningUserIds();
                processInstancePriority = schedule.getProcessInstancePriority();
            }
        }

        //启动实例Id
        String processInstanceId = "";

        // 启动包含扫描目录节点的流程定义，如果不存在扫描目录的定时任务，则创建
        if (processDefinition.getProcessDefinitionJson().contains(TaskType.MONITORDIRECTORY.name())) {
            //此处是只执行一次, 在流程启用处，需要增加定时扫描目录
            String jobName = QuartzExecutors.buildJobName(processDefinition.getId());
            String jobGroupName = QuartzExecutors.buildJobGroupName(processDefinition.getProjectId());

            Map<String, Object> dataMap = new HashMap<>(3);
            dataMap.put(Constants.PROJECT_ID, processDefinition.getProjectId());
            dataMap.put(Constants.PROCESS_DEFINITION_ID, processDefinition.getId());

            QuartzExecutors.getInstance().executeOnceJob(MonitorDirectoryScheduleJob.class, jobName, jobGroupName, dataMap);

        } else {
            // step3:创建CMD
            Command command = this.createCommand(commandType, processDefinitionId, taskDependType, failureStrategy, startNodeList,
                    cronTime, warningType, "default", warningUserIds, runMode, processInstancePriority, workerGroup);

            if (command == null) {
                logger.error("start process definition failed, id={}", processDefinitionId);
                throw new BusinessException(String.format("start process definition failed, id=%s", processDefinitionId));
            }

        }
        //等待流程实例启动获取流程实例ID, 判断5秒内
        int count = 0;
        while (true && count <= 10) {
            processInstanceId = RedisCacheHelper.getInstance().get("bdp:dds:processDefinitionId:" + processDefinitionId);
            if (StringUtils.isBlank(processInstanceId)) {
                try {
                    Thread.sleep(500);
                    count++;
                    logger.info("get process instance id try times: {}--{}", processInstanceId, count);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            } else {
                RedisCacheHelper.getInstance().del("bdp:dds:processDefinitionId:" + processDefinitionId);
                break;
            }
        }
        logger.info("start process definition success, id={}, instanceId={}", processDefinitionId, processInstanceId);
        //记录日志
        String logTitle = "运行任务：【" + processDefinition.getName() + "】";
        BdpLogUtil.log("数据计算模块", logTitle, PlatformConstant.OpType.update);
        return processInstanceId;
    }

    /**
     * do action to process instance：pause, stop, repeat, recover from pause, recover from stop
     *
     * @param processDefinitionId 传入processDefinitionId，然后根据processDefinitionId查找最新processInstanceId
     * @param processInstanceId   流程实例ID
     * @param executeType         execute type
     */
    public void execute(String processDefinitionId, String processInstanceId, ExecuteType executeType) {
        // step1:检查master服务是否存在
        if (!checkMasterExists()) {
            logger.error("master service doesn't exist");
            throw new BusinessException("master service doesn't exist");
        }

        // step2:查询流程实例是否存在
        ProcessInstance processInstance;
        if (StringUtils.isNotBlank(processDefinitionId)) {
            //DCS调用接口，根据流程定义ID获取启动时间最近的流程实例
            List<ProcessInstance> processInstances = processInstanceService.queryByProcessDefineId(processDefinitionId, 1);
            if (CollectionUtils.isEmpty(processInstances)) {
                logger.error("process instance does not exist, processDefinitionId={}", processDefinitionId);
                throw new BusinessException("process instance does not exist");
            }
            processInstance = processInstances.get(0);
        } else {
            processInstance = processInstanceService.queryProcessInstanceById(null, processInstanceId);
        }

        if (processInstance == null) {
            logger.error("process instance {} does not exist", processInstanceId);
            throw new BusinessException("process instance does not exist");
        }

        processInstanceId = processInstance.getId();
        processDefinitionId = processInstance.getProcessDefinitionId();

        // step3:检查流程定义合法性
        ProcessDefinition processDefinition = processDefinitionMapper.selectByPrimaryKey(processDefinitionId);
        if (executeType != ExecuteType.STOP && executeType != ExecuteType.PAUSE && executeType != ExecuteType.REPEAT_RUNNING) {
            if (!checkProcessDefinitionValid(processDefinition, processInstanceId)) {
                throw new BusinessException(String.format("process definition is invalid, id=%s", processDefinitionId));
            }
        }

        // step4:检查执行类型合法性
        if (!checkExecuteType(processInstance, executeType)) {
            logger.error("check execute type failed, processInsId={}, executeType={}", processInstanceId, executeType);
            throw new BusinessException(String.format("流程实例%s状态%s,不允许执行%s操作,请重新查看状态",
                    processInstance.getName(), ExecutionStatus.of(processInstance.getState()).getDescp(), executeType.toString()));
        }

        // step5:根据执行类型分别处理
        boolean result = true;
        switch (executeType) {
            case REPEAT_RUNNING:
                result = insertCommand("default", processInstanceId, processDefinitionId, CommandType.REPEAT_RUNNING.getCode());
                break;
            case RECOVER_SUSPENDED_PROCESS:
                result = insertCommand("default", processInstanceId, processDefinitionId, CommandType.RECOVER_SUSPENDED_PROCESS.getCode());
                break;
            case START_FAILURE_TASK_PROCESS:
                result = insertCommand("default", processInstanceId, processDefinitionId, CommandType.START_FAILURE_TASK_PROCESS.getCode());
                break;
            case STOP:
                ExecutionStatus status = ExecutionStatus.of(processInstance.getState());
                if (processInstance.getState() == ExecutionStatus.READY_STOP.getCode() || status.typeIsFinished()) {
                    logger.error("流程实例{}状态已经是{}", processInstance.getName(), processInstance.getState());
                    throw new BusinessException(String.format("流程实例%s状态已经是%s", processInstance.getName(), processInstance.getState()));
                } else {
                    result = updateProcessInstancePrepare(processInstance, CommandType.STOP.getCode(), ExecutionStatus.READY_STOP.getCode());
                }
                break;
            case PAUSE:
                if (processInstance.getState() == ExecutionStatus.READY_PAUSE.getCode()) {
                    logger.error("流程实例{}状态已经是{}", processInstance.getName(), processInstance.getState());
                    throw new BusinessException(String.format("流程实例%s状态已经是%s", processInstance.getName(), processInstance.getState()));
                } else {
                    result = updateProcessInstancePrepare(processInstance, CommandType.PAUSE.getCode(), ExecutionStatus.READY_PAUSE.getCode());
                }
                break;
            default:
                logger.error("流程实例{}执行类型{}非法", processInstanceId, executeType);
                throw new BusinessException(String.format("流程实例%s执行类型%s非法", processInstanceId, executeType));
        }

        if (!result) {
            logger.error("执行流程实例失败, id={}, executeType={}", processInstanceId, executeType);
            throw new BusinessException(String.format("执行流程实例失败, id=%s, executeType=%s", processInstanceId, executeType));
        }
    }


}
